package webtrans

import (
	"bytes"
	"encoding/base64"
	"encoding/json"
	"gitee.com/DreamPublic/signal/pkg/proto"
	"github.com/pion/webrtc/v3"
	"go.uber.org/zap"
	"io"
	"io/ioutil"
	"net/http"
	"sync"
)

type Session struct {
	id     string
	cbLock sync.RWMutex
	// header - []proto.SessionCallback
	onCbMap sync.Map
	signal  *Signal
	log     *zap.Logger
	pc      *webrtc.PeerConnection
	dc      *webrtc.DataChannel
}

func (s *Session) ID() string {
	return s.id
}

func (s *Session) On(header string, cb proto.SessionCallback) {
	s.cbLock.Lock()
	defer s.cbLock.Unlock()
	cbsRaw, ok := s.onCbMap.Load(header)
	if !ok {
		cbsRaw = make([]proto.SessionCallback, 0)
	}
	cbs := append(cbsRaw.([]proto.SessionCallback), cb)
	s.onCbMap.Store(header, cbs)
}

func (s *Session) Emit(header string, body string) error {
	msgStruct := proto.NewMsgStructFromParam(header, body)
	writeErr := s.dc.Send(msgStruct.ToBytes())
	if nil != writeErr {
		s.log.Error("发送返回数据失败", zap.Error(writeErr),
			zap.String("header", header),
			zap.String("body", body))
		return writeErr
	}
	return nil
}

func (s *Session) Close() {
	_ = s.pc.Close()
	s.signal.fireCloseCb(s)
	s.log.Info("连接断开")
}

func NewSession(sessionId string, signal *Signal) *Session {
	return &Session{
		id:      sessionId,
		cbLock:  sync.RWMutex{},
		onCbMap: sync.Map{},
		signal:  signal,
		log:     zap.L().With(zap.String("sessionId", sessionId)),
		pc:      nil,
	}
}

func (s *Session) fireOnCb(msg *proto.MsgStruct) {
	header := msg.GetHeader()
	cbsRaw, ok := s.onCbMap.Load(header)
	if ok {
		for _, cb := range cbsRaw.([]proto.SessionCallback) {
			go cb(proto.SessionCbStruct{
				Session:   s,
				MsgStruct: msg,
				SessionId: s.id,
				Msg:       msg.GetBody(),
			})
		}
	}
	cbsRaw, ok = s.onCbMap.Load("*")
	if ok {
		for _, cb := range cbsRaw.([]proto.SessionCallback) {
			go cb(proto.SessionCbStruct{
				Session:   s,
				MsgStruct: msg,
				SessionId: s.id,
				Msg:       msg.GetBody(),
			})
		}
	}
}

func (s *Session) HandleHTTP(w http.ResponseWriter, r *http.Request) {
	log := s.log

	if !s.signal.fireAuthCb(proto.AuthCbStruct{
		Path:   r.RequestURI,
		Origin: r.Header.Get("Origin"),
	}) {
		return
	}

	offerBytes, readOfferErr := ioutil.ReadAll(r.Body)
	if nil != readOfferErr {
		log.Error("读取offer失败", zap.Error(readOfferErr))
		return
	}
	decodeOfferBytes, decodeOfferErr := base64.StdEncoding.DecodeString(string(offerBytes))
	if nil != decodeOfferErr {
		log.Error("解析offer失败", zap.Error(readOfferErr))
		return
	}
	var offer webrtc.SessionDescription
	decodeOfferJsonErr := json.Unmarshal(decodeOfferBytes, &offer)
	if nil != decodeOfferJsonErr {
		log.Error("解析offer json失败", zap.Error(decodeOfferJsonErr))
		return
	}
	pc, newPcErr := s.signal.webrtcAPI.NewPeerConnection(webrtc.Configuration{
		ICEServers: []webrtc.ICEServer{
			{},
		},
	})
	if nil != newPcErr {
		log.Error("创建PC失败", zap.Error(newPcErr))
		return
	}
	setRemoteSdpErr := pc.SetRemoteDescription(offer)
	if nil != setRemoteSdpErr {
		log.Error("set remote sdp失败", zap.Error(setRemoteSdpErr))
		return
	}
	answer, createAnswerErr := pc.CreateAnswer(nil)
	if nil != createAnswerErr {
		log.Error("create answer失败", zap.Error(createAnswerErr))
		return
	}
	setLocalSdpErr := pc.SetLocalDescription(answer)
	if nil != setLocalSdpErr {
		log.Error("set local sdp失败", zap.Error(setLocalSdpErr))
		return
	}
	// 等待收集结束
	<-webrtc.GatheringCompletePromise(pc)

	answerJsonBytes, encodeAnswerJsonErr := json.Marshal(pc.LocalDescription())
	if nil != encodeAnswerJsonErr {
		log.Error("encode answer json失败", zap.Error(encodeAnswerJsonErr))
		return
	}
	answerBytes := []byte(base64.StdEncoding.EncodeToString(answerJsonBytes))
	w.WriteHeader(200)
	_, writeAnswerErr := w.Write(answerBytes)
	if nil != writeAnswerErr {
		log.Error("http write answer失败", zap.Error(writeAnswerErr))
		return
	}

	// 设置回调
	pc.OnDataChannel(func(channel *webrtc.DataChannel) {
		if nil != s.dc {
			// 只允许打开一个datachannel
			_ = channel.Close()
		}

		channel.OnOpen(func() {
			s.dc = channel
			s.signal.fireConnectCb(s)
		})

		channel.OnMessage(func(msg webrtc.DataChannelMessage) {
			// 读取数据
			lenHeaderBytes := make([]byte, 4)
			lenBodyBytes := make([]byte, 4)

			reader := bytes.NewReader(msg.Data)
			// 读取 header 长度
			_, readDataErr := io.ReadFull(reader, lenHeaderBytes)
			if nil != readDataErr {
				log.Error("read datachannel data error", zap.Error(readDataErr))
				return
			}
			// 读取 body 长度
			_, readDataErr = io.ReadFull(reader, lenBodyBytes)
			if nil != readDataErr {
				log.Error("read datachannel data error", zap.Error(readDataErr))
				return
			}

			lenHeader := proto.BytesToInt(lenHeaderBytes)
			lenBody := proto.BytesToInt(lenBodyBytes)

			leftBuf := make([]byte, lenHeader+lenBody)
			_, readDataErr = io.ReadFull(reader, leftBuf)
			if nil != readDataErr {
				log.Error("read datachannel data error", zap.Error(readDataErr))
				return
			}
			totalBuf := bytes.Buffer{}
			totalBuf.Write(lenHeaderBytes)
			totalBuf.Write(lenBodyBytes)
			totalBuf.Write(leftBuf)

			// 解析数据包
			structFromBytes := proto.NewMsgStructFromBytes(totalBuf.Bytes())
			if nil != structFromBytes {
				s.fireOnCb(structFromBytes)
			}
		})

		channel.OnClose(func() {
			s.Close()
		})
	})
	s.pc = pc
}
